package nl.justobjects.pushlet.core;

import nl.justobjects.pushlet.util.PushletException;

import java.util.LinkedList;
import java.util.concurrent.ConcurrentHashMap;

/**
 * IDEA 13
 * Created by xnkl on 14-3-22.
 */
public class SubscriberManager {
    public static final Session[] SESSIONS = new Session[0];
    private static SubscriberManager instance;

    private ConcurrentHashMap<String, Session[]> subscriberInfoCache;
    private ConcurrentHashMap<String, LinkedList<Session>> subscriberInfoMap;
    private ConcurrentHashMap<String, Subscription> subscriptionInfoMap;

    static {
        instance = new SubscriberManager();
    }

    private SubscriberManager() {

    }

    public void start() {
    	subscriberInfoCache = new ConcurrentHashMap<String, Session[]>();
        subscriberInfoMap = new ConcurrentHashMap<String, LinkedList<Session>>();
        subscriptionInfoMap = new ConcurrentHashMap<String, Subscription>();
    }

    public void stop() {
    	subscriberInfoCache.clear();
        subscriberInfoMap.clear();
        subscriptionInfoMap.clear();
    }

    public static SubscriberManager getInstance() {
        return instance;
    }

    public Subscription addSubscriberInfo(String subject, String label, Session session) throws PushletException {
        String compositeSubject = Subscription.compositeSubject(subject, label);
        Subscription subscription;
        LinkedList<Session> sessions;

        synchronized (compositeSubject.intern()) {
            if ((sessions = subscriberInfoMap.get(compositeSubject)) == null) {
                sessions = new LinkedList<Session>();
                subscriberInfoMap.put(compositeSubject, sessions);
                subscription = Subscription.create(subject, label);
                subscriptionInfoMap.put(compositeSubject, subscription);
            } else {
                subscription = subscriptionInfoMap.get(compositeSubject);
            }
            sessions.add(session);
            subscriberInfoCache.put(compositeSubject, SESSIONS);
        }

        return subscription;
    }

    public Session[] getSubscriberInfo(String subject, String label) {
        String compositeSubject = Subscription.compositeSubject(subject, label);
    	LinkedList<Session> sessions;
    	Session[] cache;
    	
		if ((sessions = subscriberInfoMap.get(compositeSubject)) == null)
			return SESSIONS;

        synchronized (compositeSubject.intern()) {
            if ((cache = subscriberInfoCache.get(compositeSubject)).length == 0) {
            	cache = sessions.toArray(SESSIONS);
            	subscriberInfoCache.put(compositeSubject, cache);
            	return cache;
            }
            return cache;
        }
    }

    public void removeSubscriberInfo(String subject, String label, Session session) {
        String compositeSubject = Subscription.compositeSubject(subject, label);
    	removeSubscriberInfo(compositeSubject, session);
    }

    public void removeSubscriberInfo(String compositeSubject, Session session) {
        LinkedList<Session> sessions;

        if ((sessions = subscriberInfoMap.get(compositeSubject)) == null)
            return;

        synchronized (compositeSubject.intern()) {
            sessions.remove(session);
            subscriberInfoCache.put(compositeSubject, SESSIONS);
            if (sessions.size() == 0) {
                subscriberInfoMap.remove(compositeSubject);
                subscriptionInfoMap.remove(compositeSubject);
                subscriberInfoCache.remove(compositeSubject);
                // 将该主题的消息缓存加入到待回收列表中
                EventCache.getInstance().recoverIn(compositeSubject);
            }
        }
    }
}
